package t20250612;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * 演示Flink中Watermark的使用
 * 
 * Watermark是Flink中处理事件时间的机制，用于处理乱序数据和延迟数据
 * 本例中模拟传感器数据，展示如何：
 * 1. 设置事件时间特性
 * 2. 分配时间戳和watermark
 * 3. 使用watermark进行窗口计算
 */
public class DataStreamWaterMark {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. 配置环境
        // 设置并行度为1，方便观察结果
        env.setParallelism(1);
        
        // 3. 创建数据源 - 模拟传感器数据
        DataStream<SensorReading> sensorData = env.addSource(new SensorSource())
                .map(new MapFunction<SensorReading, SensorReading>() {
                    @Override
                    public SensorReading map(SensorReading value) throws Exception {
                        // 打印原始数据，包括事件时间
                        System.out.println("接收到数据: " + value);
                        return value;
                    }
                });
        
        // 4. 分配时间戳和watermark
        // 使用WatermarkStrategy为数据流分配时间戳和watermark
        // 这里使用有界无序水印，允许数据最大延迟3秒
        DataStream<SensorReading> withTimestampsAndWatermarks = sensorData
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((event, timestamp) -> event.timestamp)
                );
        
        // 5. 应用窗口操作和聚合
        // 使用5秒的滚动窗口，计算每个传感器在窗口内的平均温度
        withTimestampsAndWatermarks
                .keyBy(reading -> reading.sensorId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new TemperatureAverager())
                .print();
        
        // 6. 执行任务
        System.out.println("开始执行Flink任务...");
        env.execute("Watermark Demo");
    }
    
    /**
     * 传感器读数类 - 包含传感器ID、温度值和事件时间戳
     */
    public static class SensorReading {
        public String sensorId;
        public double temperature;
        public long timestamp;
        
        public SensorReading() {}
        
        public SensorReading(String sensorId, double temperature, long timestamp) {
            this.sensorId = sensorId;
            this.temperature = temperature;
            this.timestamp = timestamp;
        }
        
        @Override
        public String toString() {
            return "SensorReading{" +
                    "sensorId='" + sensorId + '\'' +
                    ", temperature=" + temperature +
                    ", timestamp=" + timestamp +
                    ", eventTime=" + Instant.ofEpochMilli(timestamp) +
                    '}';
        }
    }
    
    /**
     * 传感器数据源 - 生成模拟的传感器数据，包括一些乱序数据
     */
    public static class SensorSource implements SourceFunction<SensorReading> {
        private boolean running = true;
        private Random random = new Random();
        private static final String[] SENSOR_IDS = {"sensor_1", "sensor_2", "sensor_3"};
        
        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            // 生成一些正常顺序的数据
            long startTime = System.currentTimeMillis();
            
            // 首先生成一批有序数据
            for (int i = 0; i < 10 && running; i++) {
                for (String sensorId : SENSOR_IDS) {
                    // 生成当前时间的数据
                    long eventTime = startTime + i * 1000; // 每秒一个数据点
                    double temperature = 20 + random.nextGaussian() * 5; // 温度在20度左右波动
                    ctx.collect(new SensorReading(sensorId, temperature, eventTime));
                }
                Thread.sleep(1000); // 每秒发送一批数据
            }
            
            // 然后生成一些乱序数据（延迟数据）
            System.out.println("现在生成一些乱序数据...");
            List<SensorReading> outOfOrderReadings = new ArrayList<>();
            
            // 创建一些延迟的数据点（时间戳比当前早）
            long currentTime = System.currentTimeMillis();
            for (int i = 0; i < 5; i++) {
                String sensorId = SENSOR_IDS[random.nextInt(SENSOR_IDS.length)];
                // 生成2-8秒前的数据
                long delay = 2000 + random.nextInt(6000);
                long eventTime = currentTime - delay;
                double temperature = 20 + random.nextGaussian() * 5;
                outOfOrderReadings.add(new SensorReading(sensorId, temperature, eventTime));
            }
            
            // 打乱顺序并发送
            java.util.Collections.shuffle(outOfOrderReadings);
            for (SensorReading reading : outOfOrderReadings) {
                if (!running) break;
                ctx.collect(reading);
                Thread.sleep(500);
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
    }
    
    /**
     * 窗口处理函数 - 计算窗口内的平均温度
     */
    public static class TemperatureAverager extends ProcessWindowFunction<
            SensorReading, 
            Tuple3<String, Double, String>, 
            String, 
            TimeWindow> {
        
        @Override
        public void process(
                String sensorId,
                ProcessWindowFunction<SensorReading, Tuple3<String, Double, String>, String, TimeWindow>.Context context,
                Iterable<SensorReading> elements,
                Collector<Tuple3<String, Double, String>> out) {
            
            int count = 0;
            double sum = 0.0;
            
            for (SensorReading reading : elements) {
                sum += reading.temperature;
                count++;
            }
            
            double avgTemp = count > 0 ? sum / count : 0.0;
            String windowInfo = "窗口: [" + 
                    Instant.ofEpochMilli(context.window().getStart()) + " 到 " + 
                    Instant.ofEpochMilli(context.window().getEnd()) + "]";
            
            out.collect(new Tuple3<>(sensorId, avgTemp, windowInfo));
        }
    }
}
